//*********************************************************
//
//    Copyright (c) Microsoft. All rights reserved.
//    This code is licensed under the Apache License, Version 2.0.
//    THIS CODE IS PROVIDED *AS IS* WITHOUT WARRANTY OF
//    ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING ANY
//    IMPLIED WARRANTIES OF FITNESS FOR A PARTICULAR
//    PURPOSE, MERCHANTABILITY, OR NON-INFRINGEMENT.
//
//*********************************************************

namespace Microsoft.Research.eResearch.WFServiceLibrary
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using SRAPI = Microsoft.Research.DataLayer;
    using Microsoft.Research.DataLayer;
    using System.Diagnostics;
    using System.Threading;
    using System.IO;
    using System.Reflection;
    using Microsoft.Research.eResearch.TridentScheduleCommon;
    using System.Collections.ObjectModel;
    using Microsoft.Research.eResearch.Common.Logging;
    using Microsoft.Research.eResearch.Common.Eventing;
    using System.Globalization;
    using Microsoft.Research.ScientificWorkflow.TridentUtilities;

    /// <summary>
    /// This class provides the service to poll the registry at regular
    /// intervals to look for new jobs that are being added to run.
    /// </summary>
    public class WFServiceScheduler : ILoggable
    {
        #region Member Data

        /// <summary>
        /// Scheduler service name.
        /// </summary>
        private const string SERVICENAME = "TridentExecutionService";

        /// <summary>
        /// Name of the instance where the Scheduler will look for
        /// jobs.
        /// </summary>
        private string hostName;


        /// <summary>
        /// If true service will show the console window.
        /// </summary>
        private bool showConsole;

        /// <summary>
        /// Handle to the registry.
        /// </summary>
        private SRAPI.Connection registryHandle = null;

        /// <summary>
        /// This variable defines the mode in which new WF's will be
        /// launched. If this is set to true then the WF is launched as
        /// part of a different process. If false then each WF is
        /// launched as a seperate thread in the context of the WF
        /// service host process.
        /// </summary>
        private bool launchWFProcess = true;

        /// <summary>
        /// Name and path of the WF Process executable.
        /// </summary>
        private string wfProcessName;

        /// <summary>
        /// Mutex object for recieveing signals from the clients.
        /// </summary>
        private Mutex[] syncObject = new Mutex[1];

        /// <summary>
        /// Common DLL Path.
        /// </summary>
        private string commonDLLPath = string.Empty;

        #endregion

        /// <summary>
        /// This is the event Handler for the terminate Event.
        /// </summary>
        /// <param name="message">Reason for Termination.</param>
        public delegate void TerminateSchedulerEvent(string message);

        /// <summary>
        /// Event for Lanch of Workflow.
        /// </summary>
        public event EventHandler<WFLauchEventArgs> OnLaunchWorkflow;

        /// <summary>
        /// Event for Terminate.
        /// </summary>
        public event TerminateSchedulerEvent OnTerminateScheduler;

        /// <summary>
        /// Member variable holding the currently configured scheduler
        /// </summary>
        private ITridentScheduler currentScheduler = null;

        /// <summary>
        /// Scheduler interface to get the current status of the job
        /// </summary>
        private ITridentScheduleStatus currentScheduleStatus = null;

        private int connectionAttempts;

        // Get or Set the instance name for which to look for running jobs
        static Machine machine = null;

        
        #region Properties

        /// <summary>
        /// Gets or sets WF Service Instance Name.
        /// </summary>
        public string HostName
        {
            get { return this.hostName; }
            set { this.hostName = value; }
        }

        /// <summary>
        /// Gets or sets to check whether to show the console while ruuning teh WF.
        /// </summary>
        public bool ShowConsole
        {
            get { return this.showConsole; }
            set { this.showConsole = value; }
        }

        /// <summary>
        /// Gets Handle to talk to the registry.
        /// </summary>
        public Connection RegistryHandle
        {
            get { return this.registryHandle; }
        }

        /// <summary>
        /// Set this property to true to launch WF processes and 
        /// false to laucn WF process as threads.
        /// </summary>
        public bool LaunchWFProcesses
        {
            get { return this.launchWFProcess; }
            set { this.launchWFProcess = value; }
        }

        /// <summary>
        /// Gets or sets Name along with the path of the WF executable.
        /// </summary>
        public string WFProcessName
        {
            get { return this.wfProcessName; }
            set { this.wfProcessName = value; }
        }

        /// <summary>
        /// 
        /// </summary>
        public int configurableReconnectAttempts { get; set; }

        /// <summary>
        /// Holds the flag wether the service can be Interactive or not
        /// </summary>
        public bool IsInteractiveExecution { get; set; }

        #endregion

        /// <summary>
        /// This method will be called by the differnt hosts for running
        /// the WF's. This method executes an infinite loop where it polls
        /// the registry to look for new jobs to run.
        /// </summary>
        public void RunWorkflows()
        {
            this.Log("Info", "Entered the RunWorkflow for {0}", this.HostName);

            //Terminate the thread and service in case the initialization is failed
            if (!this.InitJobRunner())
            {
                return;
            }

            //Terminate the thread/service in case the init scheduler is failed
            if (!this.InitializeScheduler())
            {
                return;
            }

            int verifyTimerCount = 0;
            while (true)
            {
                verifyTimerCount++;
                try
                {
                    this.ReOpenRegistryConnection();

                    this.UpdateNodeStatus();

                    this.ProcessCurrentJobs();
                    
                    //Do the verification only once in 25 times 
                    if (0 == verifyTimerCount % 25)
                    {
                        verifyTimerCount = 0;
                        this.VerifyRunningJobs();                       
                    }
                }
                catch (ThreadAbortException exp)
                {
                    this.Log(
                        "Error",
                        "Registry connection stopped {0} in WFServiceScheduler due to thread aborted: {1}",
                        this.HostName,
                        exp.Message);
                    break;
                }
                catch (Exception exp)
                {
                    connectionAttempts++;

                    this.Log(
                        "Error",
                        "Could not Re-Open Registry for {0} in WFServiceScheduler due to : {1} Retrying attempt : {2}",
                        this.HostName,
                        exp.Message,
                        connectionAttempts);
                    //In case of connection error wait for 10 seconds before connection
                    Thread.Sleep(5000);
                }
                Thread.Sleep(5000);
            }
            this.Log("Error", "Job runner thread is stopped", this.HostName);
        }

        /// <summary>
        /// Initialize the Job runner
        /// </summary>
        /// <returns></returns>
        private bool InitJobRunner()
        {
            bool initSuccess = false;
            try
            {
                // Let the SQL backend available to connect.
                this.OpenRegistry();

                if (!LoadMachine())
                {
                    throw new Exception("Machine load failed");
                }
                initSuccess = true;
            }
            catch (ThreadAbortException exp)
            {
                this.Log("Error", "Thread aborted  for {0} during Registry connection in WFServiceScheduler : {1}", this.HostName, exp.Message);                
            }
            catch (Exception exp)
            {
                if (null != this.OnTerminateScheduler)
                {
                    this.OnTerminateScheduler("Could Not Open Registry for " + this.HostName + " in WFServiceScheduler due to :" + exp.Message);                    
                }
                else
                {
                    this.Log("Error", "Could Not Open Registry for {0} in WFServiceScheduler due to : {1}", this.HostName, exp.Message);                    
                }
            }
            return initSuccess;
        }
        /// <summary>
        /// 
        /// </summary>
        /// <returns></returns>
        private bool InitializeScheduler()
        {
            bool initSuccess = false;
            try
            {

                //Load the configured scheduler from current folder 
                //LoadScheduler will throw in case of error, initialize the class otherwise
                LoadScheduler();
                //initialize the scheduler 
                currentScheduler.Initialize(this.wfProcessName,
                                            this.RegistryHandle,
                                            this.wfProcess_Exited,
                                            this.OnLaunchWorkflow);
                machine.Refresh();
                currentScheduler.LogMessageReceived += new EventHandler<LogEventArgs>(OnLogMessageReceived);
                initSuccess = true;
            }
            catch (Exception exp)
            {
                if (null != this.OnTerminateScheduler)
                {
                    this.OnTerminateScheduler("Load and initialize the scheduler on " + this.hostName +
                        " for WFServiceScheduler failed due to: " + exp.Message);
                }
                else
                {
                    this.Log(
                        "Error",
                        "Load and initialize the scheduler on {0} for WFServiceScheduler failed due to: {1}",
                        this.hostName,
                        exp.Message);
                    
                }
            }
            return initSuccess;
        }
        /// <summary>
        /// Process the current jobs with the schedulers
        /// </summary>
        private void ProcessCurrentJobs()
        {
            try
            {
                //Update the registry handle once get reconnected
                this.currentScheduler.UpdateRegistryConnection(this.registryHandle);

                List<Job> pendingJobs = ThreadBase.GetJobsToRun(machine);

                if (null != pendingJobs && 0 != pendingJobs.Count)
                {
                    try
                    {
                        this.Log("Info", "Found {0} jobs to run.", pendingJobs.Count.ToString());
                        //Refresh the machine object to get all the changes from management studio
                        machine.Refresh();
                        //Call the current scheduler to process the jobs list
                        currentScheduler.ProcessJobs(this.showConsole, new Collection<Job>(pendingJobs), machine);
                    }
                    catch (Exception exp)
                    {
                        this.Log("Error",
                            "Calling the current scheduler to process the job list on {0} for WFServiceScheduler failed due to: {1}",
                            this.hostName,
                            exp.Message);
                    }
                }
            }
            catch (RecordNotFoundException recordNotFound)
            {
                this.Log("Error", recordNotFound.Message);
            }
        }
        /// <summary>
        /// Verifies the running jobs to make sure the status are correct
        /// Updates the running status to Aborted in case if the actual job is not running and the registry status is running
        /// </summary>
        private void VerifyRunningJobs()
        {
            try
            {
                //Get the jobs with running/Startpending status
                List<Job> runningJobs = ThreadBase.GetJobsToVerify(machine, registryHandle);

                if (null != runningJobs && 0 != runningJobs.Count)
                {
                    this.Log("Info", "Found {0} jobs running.", runningJobs.Count.ToString());

                    foreach (Job job in runningJobs)
                    {
                        try
                        {
                            JobStatus status = currentScheduleStatus.GetScheduleStatus(job);
                            //Mismatch in status, update it in job
                            if (status == JobStatus.Aborted)
                            {
                                job.Refresh();
                                if (job.Status == JobStatus.Running ||
                                    job.Status == JobStatus.StartPending)
                                {
                                    this.Log("Info", "Changed status to aborted. Jobid : {0}", ((IObject)job).ID.ToString());
                                    job.ErrorMessage = "Job aborted due to unknown reason";
                                    job.Status = JobStatus.Aborted;
                                }
                            }
                        }
                        catch (Exception exp)
                        {
                            this.Log("Error",
                                "Calling the current scheduler to get the job status on {0} failed due to: {1}",
                                this.hostName,
                                exp.Message);
                        }
                    }
                }
            }
            catch (RecordNotFoundException)
            {

            }
        }

        /// <summary>
        /// Creates the machine.
        /// </summary>
        /// <returns></returns>
        private Machine CreateMachine()
        {
            Machine machine = null;
            try
            {
                machine = ThreadBase.CreateMachine(this.hostName, this.RegistryHandle);

                if (this.IsInteractiveExecution)
                {
                    machine.IsInteractiveExecution = this.IsInteractiveExecution;
                }

            }
            catch (TridentCustomException exp)
            {
                this.Log("Error", exp.Message);
            }
            return machine;
        }


        /// <summary>
        /// Updates the node status.
        /// </summary>
        private void UpdateNodeStatus()
        {
            Machine currentNode = ThreadBase.LoadMachine(this.HostName, this.RegistryHandle, this.IsInteractiveExecution);
            ThreadBase.UpdateExecutionServiceStatus(currentNode);
        }

        #region Private Methods

        /// <summary>
        /// This method created the mutex for talking to WF client processes.
        /// </summary>
        private void CreateMutex()
        {
            bool createdNew;
            try
            {
                this.syncObject[0] = new Mutex(true, this.hostName, out createdNew);

                if (false == createdNew)
                {
                    this.Log("Error", "Mutex alread exists");
                }

                this.Log("Info", "Mutex Created Successfully");
            }
            catch (Exception exp)
            {
                if (null != this.OnTerminateScheduler)
                {
                    this.OnTerminateScheduler("Could Not open Registry Due to :" + exp.Message);
                }
                else
                {
                    this.Log("Error", "Could Not open Registry Due to : {0}", exp.Message);
                    return;
                }
            }
        }

        void wfProcess_Exited(object sender, EventArgs e)
        {
            Process currProcess = sender as Process;

            if (null != currProcess)
            {
                try
                {
                    currProcess.Dispose();

                    // Here the process exited handler will take some time to dipose all the resources.
                    // So we are making the thread to wait for a sec to make sure all the resources are released.
                    Thread.Sleep(1000);

                    ProcessStartInfo pi = currProcess.StartInfo;

                    // get the job id from arguments of the Process.
                    string jobID = pi.Arguments.Split(" ".ToCharArray())[1];

                    string directoryPath = Path.Combine(Path.GetTempPath(), jobID);

                    if (Directory.Exists(directoryPath))
                    {
                        Directory.Delete(directoryPath, true);
                    }
                }
                catch (Exception exp)
                {
                    // ignore the exception
                    this.Log("Error",
                        "Process exited handler on {0} for WFServiceScheduler failed due to: {1}",
                        this.hostName,
                        exp.Message);
                }
            }
        }

        void OnLogMessageReceived(object sender, LogEventArgs e)
        {
            this.Log(e.Level, e.Message);
        }

        /// <summary>
        /// Open the Registry from config file whihc has Default Connection Set.
        /// </summary>
        private void OpenRegistry()
        {

            if (string.IsNullOrEmpty(this.hostName))
            {
                if (null != this.OnTerminateScheduler)
                {
                    this.OnTerminateScheduler("Agruments not in Correct Format.");
                }
                else
                {
                    this.Log("Error", "Agruments not in Correct Format.");
                    return;
                }
            }

            if (null == ThreadBase.StaticRegistryHandle && null == this.RegistryHandle)
            {
                try
                {
                    BackendStorageException connFailure;
                    SR_Connection.Init();
                    ConnectionManager mgr = ConnectionManager.CreateForAgent(
                    ConnectionManager.CancelBehavior.ThrowException, ConnectionManager.Agent.ExecutionService);
                    //Try to reconnect to registry till we get connected
                    while (!mgr.TestConnection(mgr.DefaultConnection, out connFailure))
                    {
                        Thread.Sleep(10000);                       
                    }

                    this.connectionAttempts = 0;
                    this.registryHandle = mgr.PickConnection(ConnectionManager.ConnectionUI.NeverShowUI);
                    ThreadBase.StaticRegistryHandle = this.RegistryHandle;
                    this.Log("Info", "Opened connection to the registry in WFServiceScheduler for {0}", this.HostName);
                }
                catch (Exception ex)
                {
                    this.Log("Error", "Error in opening the connection {0}", ex.Message);
                }
            }
        }

        /// <summary>
        /// This function is used to check and 
        /// reopen the registry connection if it is closed.
        /// </summary>
        private void ReOpenRegistryConnection()
        {
            if (null == this.RegistryHandle || !this.RegistryHandle.Alive)
            {
                ConnectionManager mgr = ConnectionManager.CreateForAgent(
                    ConnectionManager.CancelBehavior.ThrowException,
                    ConnectionManager.Agent.ExecutionService);

                this.registryHandle = mgr.PickConnection(ConnectionManager.ConnectionUI.NeverShowUI);
                ThreadBase.StaticRegistryHandle = this.RegistryHandle;
                //Reload the machine after SQL Reconnection
                if (!LoadMachine())
                {
                    throw new Exception("Machine load failed");
                }
                connectionAttempts = 0;
            }
        }

        /// <summary>
        /// Function loads the configured scheduler library from the same folder, throws exception in case of 
        /// any error or multiple schedulers present in the folder
        /// </summary>
        private void LoadScheduler()
        {
            DirectoryInfo dirInfo = new DirectoryInfo(Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location));
            //Enumerate the engine dlls available in the system
            FileInfo[] fileinfos = dirInfo.GetFiles("Trident.Scheduler*.dll");
            //Check whether we have more than on engine, throw error in case of multiple
            if (fileinfos.Count() != 1)
            {
                this.Log("Error", "Trident Scheduler library not configured or more than one Schedulers present in the " +
                                              "configured folder. Please contact administrator to configure Trident scheduler library");
                throw new Exception("Trident Scheduler library not configured or more than one Schedulers present in the " +
                                    "configured folder. Please contact administrator to configure Trident scheduler library");
            }
            //First file info element is guaranteed becuase of the above checking
            Assembly asm = Assembly.LoadFile(fileinfos[0].FullName);
            Type[] asmTypes = asm.GetTypes();
            foreach (Type t in asmTypes)
            {
                Type[] ifaces = t.GetInterfaces();
                if (null != ifaces && ifaces.Contains(typeof(ITridentScheduler)))
                {
                    //Found the class which has Schedule engine implementation, create an object and keep it
                    ITridentScheduler scheduler = (ITridentScheduler)Activator.CreateInstance(t);
                    currentScheduler = scheduler;

                    ITridentScheduleStatus scheduleStatus = (ITridentScheduleStatus)scheduler;
                    currentScheduleStatus = scheduleStatus;
                    break;
                }
            }
            //Check whether we got a proper interface
            if (null == currentScheduler)
            {
                this.Log("Error", "Scheduler library is not configured properly." +
                                      "Please contact administrator to configure scheduler library.");
                throw new Exception("Scheduler library is not configured properly." +
                                      "Please contact administrator to configure scheduler library");
            }
        }
        /// <summary>
        /// Load machine object for this executor
        /// </summary>
        /// <returns></returns>
        private bool LoadMachine()
        {
            bool machineLoaded = false;
            // Get or Set the instance name for which to look for running jobs
            try
            {
                machine = ThreadBase.LoadMachine(this.HostName, this.RegistryHandle, this.IsInteractiveExecution);
                machineLoaded = true;
            }
            catch (RecordNotFoundException recordNotFound)
            {
                this.Log("Error", "Machine Instance {0} was not found in WFServiceScheduler due to : {1}", this.hostName, recordNotFound.Message);
                this.Log("Error", "Creating Machine Instance {0} in the Registry for WFServiceScheduler.", this.hostName);

                machine = this.CreateMachine();
                machineLoaded = true;
            }
            catch (TridentCustomException exp)
            {
                this.Log("Error", exp.Message);
            }
            catch (Exception exp)
            {
                this.Log("Error", exp.Message);
            }
            if (null == machine)
            {
                if (null != this.OnTerminateScheduler)
                {
                    this.OnTerminateScheduler("Machine Instance " + this.hostName + " cannot be loaded in WFServiceScheduler.");
                }
                else
                {
                    this.Log("Error", "Machine Instance {0} cannot be loaded in WFServiceScheduler.", this.hostName);
                }
            }
            return machineLoaded;
        }
        #endregion

        #region ILoggable Members

        /// <summary>
        /// Fired when a log message is received.
        /// </summary>
        public event EventHandler<LogEventArgs> LogMessageReceived;

        /// <summary>
        /// Raises the LogMessageReceived event.
        /// </summary>
        /// <param name="level">The log level.</param>
        /// <param name="message">The log message.</param>
        protected virtual void OnLogMessageReceived(string level, string message)
        {
            this.LogMessageReceived.Fire(this, new LogEventArgs(level, message));
        }

        /// <summary>
        /// Writes the log message.
        /// </summary>
        /// <param name="level">The log level.</param>
        /// <param name="message">Message string to be logged.</param>
        private void Log(string level, string message)
        {
            this.OnLogMessageReceived(level, String.Format(CultureInfo.CurrentUICulture, "WindowsTaskCreator: {0}", message));
        }

        /// <summary>
        /// Writes the log message.
        /// </summary>
        /// <param name="level">The log level.</param>
        /// <param name="format">Format of the message.</param>
        /// <param name="args">Arguments to be formatted.</param>
        private void Log(string level, string format, params object[] args)
        {
            this.Log(level, String.Format(CultureInfo.CurrentUICulture, format, args));
        }

        #endregion
    }
}
